package com.tang.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * DataSet API 实现 wordCount（不推荐）
 * 为什么不推荐？因为workCount已经被标记为过时的了，在1.12实现了真正的流批一体。
 * 读取文件，本质还是DataSet,https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/dataset/overview/
 *
 * @author tang
 * @since 2023/5/29 10:54
 */
public class WordCountBatchDemo {

    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取数据，从文件中读取
        DataSource<String> lineDs = env.readTextFile("input/word.txt");

        // 3. 切分转换
        FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lineDs.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                // 3.1 按照空格切分单次
                String[] words = value.split(" ");
                for (String word : words) {
                    Tuple2<String, Integer> of = Tuple2.of(word, 1);
                    collector.collect(of);
                }
            }

        });
        
        // 4. 按照word分组
        UnsortedGrouping<Tuple2<String, Integer>> wordAndOneGroupBy = wordAndOne.groupBy(0);

        // 5. 各分组内聚合
        AggregateOperator<Tuple2<String, Integer>> sum = wordAndOneGroupBy.sum(1); // 1是位置，表示第二个元素

        // 6. 输出
        sum.print();

    }

}
